package com.baiduspring.elasticsearch;

import com.baiduspring.elasticsearch.connect.ClientSingle;
import com.baiduspring.elasticsearch.error.ElasticSearchException;
import com.baiduspring.elasticsearch.operation.*;
import com.baiduspring.utils.io.FileIO;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.log4j.Log4j2;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * ES脚手架
 *
 * @author xiongyan
 */
@Log4j2
public class ElasticsearchHandle {
    private TransportClient client;

    public void scriptUpdateFieldValue(String index, String type, String id, String script) {
        try {
            UpdateRequest updateRequest = new UpdateRequest(index, type, id).script(new Script(script));
            client.update(updateRequest).get();
        } catch (ElasticSearchException e) {
            log.error("脚本执行失败，错误：{}", e);
            throw e;
        } catch (InterruptedException e) {
            log.error("脚本执行失败，错误：{}", e);
            throw new ElasticSearchException("脚本执行失败");
        } catch (ExecutionException e) {
            log.error("脚本执行失败，错误：{}", e);
            throw new ElasticSearchException("脚本执行失败");
        }
    }

    /**
     * 更新索引
     *
     * @param index 索引名称
     * @param type  索引类型
     * @param id    资源ID
     * @param doc   更新字段值
     * @return 返回更新结果 true：成功   false：失败
     */
    public boolean updateIndex(String index, String type, String id, Doc<String, Object> doc) {
        XContentBuilder json;
        try {
            json = XContentFactory.jsonBuilder().startObject();
            for (Map.Entry<String, Object> entry : doc.entrySet()) {
                json.field(entry.getKey(), entry.getValue());
            }
            json.endObject();
        } catch (IOException e) {
            log.error("数据设置错误：{}", e);
            throw new ElasticSearchException("数据设置错误");
        }

        UpdateRequest updateRequest = new UpdateRequest(index, type, id).doc(json);

        try {
            UpdateResponse response = client.update(updateRequest).get();
            Integer ok = 200;

            return ok.equals(response.status().getStatus());
        } catch (InterruptedException | ExecutionException e) {
            log.error("索引修改错误：{}", e);
            throw new ElasticSearchException("索引修改错误");
        }
    }

    /**
     * 索引文档
     * 适合于批量创建
     *
     * @param indexName           索引名
     * @param type                类型
     * @param mappingJsonFilePath mapping映射模板
     * @param docs                被索引的文档
     */
    public void indexDocs(String indexName, String type, String mappingJsonFilePath, List<Doc<String, Object>> docs) {
        createIndex(indexName, type, mappingJsonFilePath, docs);
    }

    /**
     * 索引文档
     * 适合于单个创建
     *
     * @param indexName           索引名
     * @param type                类型
     * @param id                  ID
     * @param mappingJsonFilePath mapping映射模板
     * @param doc                 被索引的文档
     */
    public void indexDocs(String indexName, String type, String id, String mappingJsonFilePath, Doc<String, Object> doc) {
        createIndex(indexName, type, id, mappingJsonFilePath, doc);
    }

    /**
     * 多字段值进行结果查询。如key1=123 and key2=456 ...
     *
     * @param input 查询参数
     * @param clazz 返回结果分局类型
     * @return 返回查询结果
     */
    public Optional<QueryResultsOutput> searchMultipleConditions(QueryInput input, QueryBuilder queryBuilder, Class clazz) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(queryBuilder);

        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(input.getIndices())
                .setTypes(input.getTypes())
                .setQuery(queryBuilder)
                .setExplain(true);

        if (Objects.nonNull(input.getCurrentPage())) {
            //分页查询开始位置
            searchRequestBuilder.setFrom(input.from())
                    //每一页的显示条数
                    .setSize(input.getPageSize());
        }

        /*
         * 设置排序字段以及排序规则
         */
        Doc<String, SortOrder> doc = input.getSortFields();
        if (MapUtils.isNotEmpty(doc)) {
            /*
             * 设置需要排序的字段
             */
            for (Map.Entry<String, SortOrder> entry : doc.entrySet()) {
                searchRequestBuilder.addSort(entry.getKey(), entry.getValue());
            }
        }

        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();

        //返回数据的字段
        Map<String, Class> outputFields = input.getOutputFields();
        Set<Map.Entry<String, Class>> entries = outputFields.entrySet();
        //保存返回数据
        List results = Lists.newArrayList();

        SearchHits hits = searchResponse.getHits();
        long total = hits.totalHits;
        resultHandle(clazz, entries, results, hits);

        QueryResultsOutput output = new QueryResultsOutput();
        output.totalPages(total, input.getPageSize());
        output.setResults(results);

        return Optional.ofNullable(output);
    }

    public Long totalMultipleConditions(QueryInput input, QueryBuilder queryBuilder) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(queryBuilder);

        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(input.getIndices())
                .setTypes(input.getTypes())
                .setQuery(queryBuilder)
                .setExplain(true);

        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        SearchHits hits = searchResponse.getHits();

        return hits.totalHits;
    }

    /**
     * 根据条件查询总条数
     *
     * @param input      入参，过滤条件
     * @param noKeyWords 标识是否根据关键字查询
     * @return 返回匹配的总数量值
     */
    public long total(QueryInput input, boolean noKeyWords) {
        SearchResponse response = noKeyWords ? searchByNoKeyWords(input) : seachByKeyWords(input);
        return response.getHits().totalHits;
    }

    /**
     * 根据条件检索数据并分页
     *
     * @param input      入参
     * @param noKeyWords 标识是否根据关键字查询
     * @param clazz      数据数据模型calss bean，如User.class
     */
    public Optional<QueryResultsOutput> search(QueryInput input, boolean noKeyWords, Class clazz) {
        QueryResultsOutput output = new QueryResultsOutput();

        //返回数据的字段
        Map<String, Class> outputFields = input.getOutputFields();
        Set<Map.Entry<String, Class>> entries = outputFields.entrySet();
        //保存返回数据
        List results = Lists.newArrayList();

        /*
         * 执行查询方法
         */
        SearchResponse response;
        if (noKeyWords) {
            response = searchByNoKeyWords(input);
        } else {
            response = search(input);
        }

        SearchHits hits = response.getHits();
        //总条数
        Long total = hits.totalHits;

        /*
         * 如果没有符合条件的数据，直接返回null
         */
        if (total.equals(0L)) {
            return Optional.empty();
        }

        if (noKeyWords) {
            resultHandle(clazz, entries, results, hits);
        } else {
            for (SearchHit hit : hits) {
                Map<String, HighlightField> highlightFieldMap = hit.getHighlightFields();

                Map<String, Object> map = Maps.newHashMap();
                map.put("id", hit.getId());

                for (Map.Entry<String, Class> entry : entries) {
                    HighlightField highlightField = highlightFieldMap.get(entry.getKey());

                    if (!Objects.isNull(highlightField)) {
                        Text[] texts = highlightField.getFragments();

                        if (texts != null && texts.length > 0) {
                            String text = texts[0].string();
                            setResults(map, entry, text);
                        }
                    } else {
                        Map<String, Object> sourceMap = hit.getSourceAsMap();
                        String text = String.valueOf(sourceMap.get(entry.getKey()));
                        setResults(map, entry, text);
                    }
                }
                Object obj = BeanUtils.map2bean(clazz, map);
                results.add(obj);
            }
        }

        output.totalPages(total, input.getPageSize());
        output.setResults(results);
        return Optional.ofNullable(output);
    }

    private void resultHandle(Class clazz, Set<Map.Entry<String, Class>> entries, List results, SearchHits hits) {
        for (SearchHit hit : hits) {
            Map<String, Object> sourceMap = hit.getSourceAsMap();

            Map<String, Object> map = Maps.newHashMap();
            entries.forEach(entry -> {
                String text = "id".equalsIgnoreCase(entry.getKey()) ? hit.getId() : String.valueOf(sourceMap.get(entry.getKey()));
                setResults(map, entry, text);
            });


            Object obj = BeanUtils.map2bean(clazz, map);
            results.add(obj);
        }
    }

    private void setResults(Map<String, Object> map, Map.Entry<String, Class> entry, String text) {
        /*
         * 数字类型
         */
        if (entry.getValue().equals(Short.class)) {
            map.put(entry.getKey(), Short.valueOf(text));
        }
        if (entry.getValue().equals(Integer.class)) {
            map.put(entry.getKey(), Integer.valueOf(text));
        }
        if (entry.getValue().equals(Long.class)) {
            map.put(entry.getKey(), Long.valueOf(text));
        }
        if (entry.getValue().equals(Float.class)) {
            map.put(entry.getKey(), Float.valueOf(text));
        }
        if (entry.getValue().equals(Double.class)) {
            map.put(entry.getKey(), Double.valueOf(text));
        }

        /**
         * 特殊类型
         */
        String s = "null";
        if (!Objects.isNull(text) && !s.equals(text) && entry.getValue().equals(Date.class)) {
            /*
             * text值格式如“2017-12-15T09:04:44.157Z”，日期转换报错：java.text.ParseException: Unparseable date: "2017-12-15T09:04:44.157Z"
             * 为了解决这个问题，故此做如下处理
             */
            text = text.replaceAll("T", " ").replaceAll("Z", "");
            map.put(entry.getKey(), DateUtils.timeToDate(text));
        }
        if (entry.getValue().equals(String.class)) {
            map.put(entry.getKey(), text);
        }
    }


    /**
     * 根据指定的ID值进行检索数据
     *
     * @param index 索引
     * @param type  类型
     * @param id    主键ID
     * @return QueryResultsOutput对象
     */
    public QueryResultsOutput searchById(String index, String type, String id, Class clazz) {
        GetResponse response = client.prepareGet(index, type, id).get();
        if (!Objects.isNull(response)) {
            Map<String, Object> map = response.getSourceAsMap();
            Object obj = BeanUtils.map2bean(clazz, map);

            QueryResultsOutput output = new QueryResultsOutput();
            output.setResult(obj);
            return output;
        }

        return null;
    }

    /**
     * 根据关键字多条件查询数据并且实现分页功能
     *
     * @param input 入参
     */
    private SearchResponse search(QueryInput input) {
        return seachByKeyWords(input);
    }

    /**
     * 根据关键字进行多条件的查询，实现分页。
     * 可以在多个Indies和types中进行检索
     *
     * @param input 入参，查询条件
     * @return 返回检索结果对象SearchResponse
     */
    private SearchResponse seachByKeyWords(QueryInput input) {
        MultiMatchQueryBuilder queryBuilder = QueryBuilders.multiMatchQuery(input.getKeywords(), input.getFilterFields());
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        queryBuilder.analyzer("ik_max_word");


        highlightBuilder.preTags("<span style=\"color:#d00;\">");
        highlightBuilder.postTags("</span>");
        /**
         * 设置需要高亮的字段
         */
        String[] highlightFields = input.getHighlightFields();
        if (highlightFields != null && highlightFields.length > 0) {
            for (String str : input.getHighlightFields()) {
                highlightBuilder.field(str);
            }
        }

        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(input.getIndices())
                //索引类型
                .setTypes(input.getTypes())
                //查询方式
                .setQuery(queryBuilder)
                //设置高亮
                .highlighter(highlightBuilder)
                //分页查询开始位置
                .setFrom(input.from())
                //每一页的显示条数
                .setSize(input.getPageSize())
                //true：按照查询匹配度排序
                .setExplain(true);
        /**
         * 设置排序字段以及排序规则
         */
        Doc<String, SortOrder> doc = input.getSortFields();
        if (doc != null && !doc.isEmpty()) {
            /*
             * 设置需要排序的字段
             */
            for (Map.Entry<String, SortOrder> entry : doc.entrySet()) {
                searchRequestBuilder.addSort(entry.getKey(), entry.getValue());
            }
        }

        return searchRequestBuilder.execute().actionGet();
    }

    /**
     * 没有按照关键字检索
     * 在指定的Indies和types中查询并实现分页
     *
     * @param input 入参，过滤条件
     * @return 返回结果集对象SearchResponse
     */
    private SearchResponse searchByNoKeyWords(QueryInput input) {
        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(input.getIndices()).setTypes(input.getTypes())
                .setExplain(true)
                .setFrom(input.from())
                .setSize(input.getPageSize());

        Doc<String, SortOrder> sortOrderDoc = input.getSortFields();
        if (sortOrderDoc != null && !sortOrderDoc.isEmpty()) {
            for (Map.Entry<String, SortOrder> doc : sortOrderDoc.entrySet()) {
                searchRequestBuilder.addSort(doc.getKey(), doc.getValue());
            }
        }

        return searchRequestBuilder.execute().actionGet();
    }

    /**
     * 对es结果值的处理
     *
     * @param clazz     入参
     * @param results   入参
     * @param key       入参
     * @param valueType 入参
     * @param map       入参
     * @param text      入参
     */
    private void setResults(Class clazz, List results, String key, Class valueType, Map<String, Object> map, String text) {
        /*
         * 数字类型
         */
        if (valueType.equals(Short.class)) {
            map.put(key, Short.valueOf(text));
        }
        if (valueType.equals(Integer.class)) {
            map.put(key, Integer.valueOf(text));
        }
        if (valueType.equals(Long.class)) {
            map.put(key, Long.valueOf(text));
        }
        if (valueType.equals(Float.class)) {
            map.put(key, Float.valueOf(text));
        }
        if (valueType.equals(Double.class)) {
            map.put(key, Double.valueOf(text));
        }

        /*
         * 特殊类型
         */
        String s = "null";
        if (!Objects.isNull(text) && !s.equals(text) && valueType.equals(Date.class)) {
            map.put(key, DateUtils.timeToDate(text));
        }
        if (valueType.equals(String.class)) {
            map.put(key, text);
        }

        Object obj = BeanUtils.map2bean(clazz, map);
        results.add(obj);
    }

    private void createIndex(String indexName, String type, String id, String mappingJsonFilePath, Doc<String, Object> doc) {
        if (StringUtils.isEmpty(indexName)) {
            throw new ElasticSearchException("索引名不能为空");
        }
        if (StringUtils.isEmpty(type)) {
            throw new ElasticSearchException("类型名不能为空");
        }
        if (Objects.isNull(doc)) {
            throw new ElasticSearchException("索引的文档数据不能为空");
        }

        /**
         * 检查并创建索引
         */
        createIndex(indexName);

        /**
         * 读取字段映射模板内容
         */
        String json = FileIO.readFileContentOfText(mappingJsonFilePath);
        PutMappingResponse mappingResponse = client.admin().indices().preparePutMapping(indexName)
                .setType(type)
                .setSource(json, XContentType.JSON)
                .get();
        if (!mappingResponse.isAcknowledged()) {
            log.error("Mapping创建失败");
            throw new ElasticSearchException("Mapping创建失败");
        }

        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
        try {
            XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().startObject();

            for (DocMap.Entry entry : doc.entrySet()) {
                String key = String.valueOf(entry.getKey());
                if ("es_id".equals(key)) {
                    continue;
                }
                xContentBuilder.field(String.valueOf(entry.getKey()), entry.getValue());
            }
            xContentBuilder.endObject();

            IndexRequestBuilder indexRequestBuilder = client.prepareIndex(indexName, type, id);
            indexRequestBuilder.setSource(xContentBuilder);
            bulkRequestBuilder.add(indexRequestBuilder);
        } catch (IOException e) {
            log.error("索引文档失败，错误：{}", e);
            throw new ElasticSearchException("索引文档失败");
        }

        BulkResponse bulkResponse = bulkRequestBuilder.get();
        if (bulkResponse.hasFailures()) {
            log.error("索引文档失败");
            throw new ElasticSearchException("索引文档失败");
        }
    }

    private void createIndex(String indexName, String type, String mappingJsonFilePath, List<Doc<String, Object>> docs) {
        if (StringUtils.isEmpty(indexName)) {
            throw new ElasticSearchException("索引名不能为空");
        }
        if (StringUtils.isEmpty(type)) {
            throw new ElasticSearchException("类型名不能为空");
        }
        if (CollectionUtils.isEmpty(docs)) {
            throw new ElasticSearchException("索引的文档数据不存在");
        }

        /*
         * 检查并创建索引
         */
        createIndex(indexName);

        /*
         * 读取字段映射模板内容
         */
        String json = FileIO.readFileContentOfText(mappingJsonFilePath);
        PutMappingResponse mappingResponse = client.admin().indices().preparePutMapping(indexName)
                .setType(type)
                .setSource(json, XContentType.JSON)
                .get();
        if (!mappingResponse.isAcknowledged()) {
            log.error("Mapping创建失败");
            throw new ElasticSearchException("Mapping创建失败");
        }

        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
        try {
            for (Doc<String, Object> doc : docs) {
                XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().startObject();

                String id = "";
                for (DocMap.Entry entry : doc.entrySet()) {
                    String key = String.valueOf(entry.getKey());
                    if ("es_id".equals(key)) {
                        id = String.valueOf(entry.getValue());
                        continue;
                    }

                    xContentBuilder.field(String.valueOf(entry.getKey()), entry.getValue());
                }
                xContentBuilder.endObject();

                IndexRequestBuilder indexRequestBuilder = client.prepareIndex(indexName, type, id);
                indexRequestBuilder.setSource(xContentBuilder);
                bulkRequestBuilder.add(indexRequestBuilder);
            }
        } catch (IOException e) {
            log.error("索引文档失败，错误：{}", e);
            throw new ElasticSearchException("索引文档失败");
        }

        BulkResponse bulkResponse = bulkRequestBuilder.get();
        if (bulkResponse.hasFailures()) {
            log.error("索引文档失败");
            throw new ElasticSearchException("索引文档失败");
        }
    }

    /**
     * 检查索引是否存在，不存在就创建
     *
     * @param indexName 索引名称
     */
    private void createIndex(String indexName) {
        try {
            EsUtils.createIndex(client, indexName);
        } catch (ElasticSearchException e) {
            log.error("索引创建失败，错误原因：{}", e);
            throw e;
        } catch (Exception e) {
            log.error("索引创建失败，错误原因：{}", e);
        }
    }

    private ElasticsearchHandle() {
        this.client = ClientSingle.openConnection();
    }

    public static ElasticsearchHandle instance() {
        return ElasticsearchHandle.GetElasticsearchHandle.get();
    }

    private final static class GetElasticsearchHandle {
        private GetElasticsearchHandle() {

        }

        private static ElasticsearchHandle get() {
            return new ElasticsearchHandle();
        }
    }
}
